package com.cmc.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.cmc.constant.MqQueueConstant;
import com.cmc.entity.User;
import com.cmc.rabbitMq.RabbitMqUtil;
import com.google.protobuf.InvalidProtocolBufferException;
import com.sun.istack.internal.NotNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author chengmingchao
 */
@Component
@Slf4j
public class CanalRunner implements CommandLineRunner {
    @Autowired
    private RabbitMqUtil rabbitMqUtil;

    /**
     * 这个run方法在容器启动完成之后执行
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        log.info("=====>启动canal监听mysql的binlog");
        CanalConnector canalConnector= CanalConnectors.newSingleConnector(
                new InetSocketAddress("192.168.98.128",11111),"example","canal","canal");
        try {
            canalConnector.connect();
            canalConnector.subscribe();
            canalConnector.rollback();
            int batchSize=1000;
            for(;;){
                Message message = canalConnector.getWithoutAck(batchSize);
                long id = message.getId();
                if (id!=-1){
                    sendToMq(message.getEntries());
                }else {
                    TimeUnit.SECONDS.sleep(1);
                }
                canalConnector.ack(id);
            }
        } catch (Exception e) {

        } finally {
            canalConnector.disconnect();
        }
    }

    /**
     * 解析并发送到mq
     * @param entries
     */
    public void sendToMq(List<CanalEntry.Entry> entries){
        for (CanalEntry.Entry entry : entries) {
            // binlog开启的模式是row，这个如果不是row的直接跳过
            if (entry.getEntryType()!= CanalEntry.EntryType.ROWDATA){
                continue;
            }
            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType== CanalEntry.EventType.DELETE){
                    User user = getColumn(rowData.getBeforeColumnsList());
                    rabbitMqUtil.send(MqQueueConstant.DELETE,JSON.toJSONString(user));
                }else if(eventType== CanalEntry.EventType.INSERT||eventType==CanalEntry.EventType.UPDATE){
                    User column = getColumn(rowData.getAfterColumnsList());
                    rabbitMqUtil.send(MqQueueConstant.INSERT_UPDATE, JSON.toJSONString(column));
                }
            }
        }
    }

    private User getColumn(@NotNull List<CanalEntry.Column> columns) {
        Map<String, Object> map = new HashMap<>();
        for (CanalEntry.Column column : columns) {
            map.put(column.getName(), column.getValue());
        }
        System.out.println("------------------>" + map.toString());
        return JSON.toJavaObject(JSON.parseObject(JSON.toJSONString(map)), User.class);
    }
}
